package com.venus.mq.client.producer;

import com.venus.mq.model.RegisterBean;
import com.venus.mq.model.Topic;
import com.venus.mq.netty.client.NettyClient;
import com.venus.mq.zk.MqRegisterBean;
import com.venus.mq.zk.ZkConfig;
import com.venus.mq.zk.ZkServiceImpl;
import lombok.Data;
import org.I0Itec.zkclient.ZkClient;

import java.net.InetSocketAddress;

/**
 * Created by mc on 18/11/22.
 */
@Data
public class Producer {

    private ProducerConfig producerConfig;

    private Topic topic;

    private MqRegisterBean mqRegisterBean;

    public Producer(ProducerConfig producerConfig, Topic topic, MqRegisterBean mqRegisterBean) {
        this.producerConfig = producerConfig;
        this.topic = topic;
        this.mqRegisterBean = mqRegisterBean;
    }

    /**
     * 初始化zookeeper注册
     */
    public  void initZk(){
        RegisterBean registerBean = new RegisterBean();
        registerBean.setAddress(producerConfig.getZkAddress());
        registerBean.setProduceOrConsumer("producer");
        registerBean.setTopicName(topic.getTopicName());
        ZkClient zkClient = ZkConfig.getInstance();
        new MqRegisterBean(new ZkServiceImpl(zkClient)).register(registerBean);
    }

    /**
     * 初始化netty 通道
     * @return
     */
    public void initNetty(){
        String zkAddress = producerConfig.getZkAddress();
        String[] zkArray = zkAddress.split(":");
        InetSocketAddress socketAddress = new InetSocketAddress(zkArray[0], Integer.parseInt(zkArray[1]));
        NettyClient.initChanel(socketAddress);
    }

    /**
     * 往broker发送topic信息
     */
    public void send(){
        NettyClient.sendMsg(topic);
    }

    /**
     * 删除producer 在zk上的节点
     */
    public  void destroyZkNode(){

    }
}
